package pipelines;

import fragment.MySQLSource;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ExecutionException;

/**
 * <pre>
 *     --PARLLELISM 1 --MYSQL_HOST 127.1 --MYSQL_PORT 3306 --MYSQL_USER root --MYSQL_PASSWORD MTIzNDU2 --MYSQL_DATABASENAME xdcx --MYSQL_TABLESNAME xdcx.mk --ACTIVE_SINKS iceberg,iceberg_delete --ICEBERG_USER root --ICEBERG_HIVE_META thrift://10.1.75.102:7004 --ICEBERG_DATABASE default --ICEBERG_TABLE_PATTERN ${table}_2 --ICEBERG_DELETED_TABLE_PATTERN ${table}_2_d
 * </pre>
 */
public class MySQLPipeline {
    private static final Logger LOGGER = LoggerFactory.getLogger(MySQLPipeline.class);

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ParameterTool pt = ParameterTool.fromArgs(args);
        long checkpointPeriod = pt.getLong("CHEECKPOINTING", 60000);
        int parallelism = pt.getInt("PARLLELISM", -1);
        Configuration configuration = new Configuration();
        StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment(configuration);
        if (parallelism > 0) {
            env.setParallelism(parallelism);
        }

        // Important!!
        env.getConfig().enableObjectReuse();
        env.enableCheckpointing(checkpointPeriod, CheckpointingMode.AT_LEAST_ONCE);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        String sourceName = new MySQLSource().createConnector(pt, tableEnv);
        JobUtils.addAllSinks(sourceName, pt, tableEnv);
    }
}
